package com.lagou.rpc;

import com.lagou.register.ICompleteCallback;
import com.lagou.register.ZkUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.pool.AbstractChannelPoolMap;
import io.netty.channel.pool.ChannelPoolHandler;
import io.netty.channel.pool.ChannelPoolMap;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.FutureListener;
import org.apache.curator.framework.CuratorFramework;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.ExecutionException;

public class ClientConnectionPool {

    /**
     * key为目标host，value为目标host的连接池
     */
    private ChannelPoolMap<InetSocketAddress, FixedChannelPool> poolMap;
    private CuratorFramework zkClient;


    public ClientConnectionPool(List<String> serverList, CuratorFramework zkClient){
        this.zkClient = zkClient;
        // 1.初始化固长连接池, 单个server 2个connection
        initPool();
        // 2.初始化server连接
        initConnection(serverList);
    }

    private void initPool(){
        EventLoopGroup group = new NioEventLoopGroup();
        final Bootstrap bootstrap = new Bootstrap();

        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true);

        poolMap = new AbstractChannelPoolMap<InetSocketAddress, FixedChannelPool>() {
            @Override
            protected FixedChannelPool newPool(InetSocketAddress key) {
                return new FixedChannelPool(bootstrap.remoteAddress(key), new ChannelPoolHandler() {
                    public void channelReleased(Channel ch) throws Exception {
                        System.out.println("channelReleased");
                        ch.writeAndFlush(Unpooled.EMPTY_BUFFER);
                    }

                    public void channelAcquired(Channel ch) throws Exception {
                        System.out.println("channelAcquired");
                    }

                    public void channelCreated(Channel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
                        pipeline.addLast(new ClientHandler());
                    }
                },2);//单个host连接池大小, 可从配置获取
            }
        };
    }

    private void initConnection(List<String> serverList) {
        if(null == serverList || serverList.isEmpty()) return;

        serverList.forEach(s -> {
            String[] arr = s.split(":");
            InetSocketAddress address = InetSocketAddress.createUnresolved(arr[0], Integer.parseInt(arr[1]));
            poolMap.get(address);
        });
    }

    public void connect(InetSocketAddress address){
        if(address == null){
            throw new RuntimeException("InetSocketAddress can not be null");
        }
        poolMap.get(address);
    }

    public void disconnect(InetSocketAddress address){
        if(address == null){
            throw new RuntimeException("InetSocketAddress can not be null");
        }
        ((AbstractChannelPoolMap)poolMap).remove(address);
    }


    /**
     * 同步获取连接并发送数据
     */
    public void write(InetSocketAddress address, final RpcRequest msg, ICompleteCallback callback) {
        if(address == null){
            throw new RuntimeException("InetSocketAddress can not be null");
        }

        // 保存客户端发起请求时间
        ZkUtil.setData(zkClient, address, String.valueOf(System.currentTimeMillis()));


        final FixedChannelPool pool = poolMap.get(address);
        try {
            Channel ch = pool.acquire().get();

            /**
             * TODO 此处是手工实现, 没找到对应api, 所以自行实现了异步回调
             */
            ClientHandler clientHandler = ch.pipeline().get(ClientHandler.class);
            //response = clientHandler.completeThenCallback();

            // 注册异步回调
            clientHandler.asyncCallback(callback);

            ch.writeAndFlush(msg);
            // 释放
            pool.release(ch);

        } catch (InterruptedException | ExecutionException e) {
            System.out.println("服务端无法连接, 可能已下线");
        }
    }

    /**
     * 异步获取连接并发送数据
     */
    public void writeAsync(InetSocketAddress address, final RpcRequest msg, ICompleteCallback callback) {
        if(address == null){
            throw new RuntimeException("InetSocketAddress can not be null");
        }

        // 保存客户端发起请求时间
        ZkUtil.setData(zkClient, address, String.valueOf(System.currentTimeMillis()));

        final FixedChannelPool pool = poolMap.get(address);

        /**
         * 异步获取连接
         */
        Future<Channel> future = pool.acquire();//.get();
        future.addListener(new FutureListener<Channel>() {

            public void operationComplete(Future<Channel> f) throws ExecutionException, InterruptedException {
                if (f.isSuccess()) {
                    Channel ch = f.get();//getNow();
                    ChannelFuture lastWriteFuture = ch.writeAndFlush(msg);

                    // Wait until all messages are flushed before closing the channel.
                    if (lastWriteFuture != null) {

                        try {
                            ClientHandler clientHandler = ch.pipeline().get(ClientHandler.class);
                            //response = clientHandler.completeThenCallback();

                            // 注册异步回调
                            clientHandler.asyncCallback(callback);

                            lastWriteFuture.sync();
                        } catch (InterruptedException e) {
                            System.out.println("服务端无法连接, 可能已下线");
                        }
                    }

                    // 释放
                    pool.release(ch);
                }
            }
        });
    }
}
